import pandas as pd
import pymysql
import time
from datetime import datetime, timedelta
from dateutil.relativedelta import relativedelta


# 设置panda显示参数
pd.set_option("display.max_rows", 20)
pd.set_option("display.max_columns", 20)
pd.set_option("display.width", 220)

mysql_config = {
    'host': 'rm-wz92r8qt89c31fn40oo.mysql.rds.aliyuncs.com',
    'port': 3306,
    'user': 'bi_rw',
    'password': 'Xjz@&283*-)',
}


class DatabaseConnector:
    def __init__(self, db_type, config):
        self.db_type = db_type
        self.config = config
        if db_type == 'presto':
            self.conn = dbapi.connect(**config)
        elif db_type == 'hive':
            self.conn = hive.connect(**config)
        elif db_type == 'mysql':
            self.conn = pymysql.connect(**config)
        self.cursor = self.conn.cursor()

    def ddl(self, cmd):
        """
        执行ddl语句,
        :param cmd: ddl操作语句
        :return: None
        """
        self.cursor.execute(cmd)
        self.conn.commit()

        # 使用presto的时候, 必须要加上fetchall()之后ddl语句才会执行, 不知道为什么
        if self.db_type == 'presto':
            self.cursor.fetchall()

        print(f"Finish running ddl: {cmd}")

    def dml(self, cmd):
        """
        执行增删改操作
            hive:   操作没有返回集, 使用fetchall会报错, 影响的行数也不可见
            presto: 结果集的类型是rows, insert语句的值是插入的行数, delete操作的值是None
        :param cmd: dml操作语句
        :return: None
        """
        self.cursor.execute(cmd)
        self.conn.commit()
        if self.db_type == 'presto':
            infected_rows = self.cursor.fetchall()[0][0]
            print(f"Finish running dml ({infected_rows} rows infected): {cmd}")
        else:
            print(f"Finish running dml: {cmd}")

    def dql(self, cmd):
        """
        执行查询操作, 需要返回查询结果集的dataframe
        :param cmd: dql操作sql
        :return: dql结果
        """
        self.cursor.execute(cmd)
        dql_desc = self.cursor.description
        dql_rows = self.cursor.fetchall()
        print(f"Finish running dql: {cmd}")
        dql_df = None
        if len(dql_rows) != 0:
            columns = [col[0] for col in dql_desc]
            result = [dict(zip(columns, row)) for row in dql_rows]
            dql_df = pd.DataFrame(result)
            dql_df.columns = columns

        return dql_df


mysql_env = DatabaseConnector('mysql', mysql_config)



run_sql = """
insert into  bi_all_in_one.dw_risk_ws_cust_score_info_record_d
with fintech_jm_s2_tag as -- 极米分
(
select 
     ri.id
    ,ri.result_code
    ,ri.source
    ,ri.ident
    ,ri.name
    ,ri.mobile
    ,replace(rv.param_value, '"','') as d_score -- 精度待调整
    ,result_text
    ,ri.created_time
    ,row_number() over(partition by ident,name,mobile,source order by created_time desc) as rn -- 如果一天同一个三要素同一调用方存在多条数据，取当天最新的记录
    from risk_ws.r_risk_info_unif_integrate_devscore_plus_fintech_jm_s2 ri 
    left join risk_ws.r_risk_value_unif_integrate_devscore_plus_fintech_jm_s2 rv on rv.info_id = ri.id and rv.param_key = 'jm_scorea2' 
    where  substr(ri.created_time,1,10) = '{check_date}'
)
,
fintech_jm_s2_user_info as 
(
    select 
     ident
    ,name
    ,mobile
    ,'unif_integrate_devscore_plus_fintech_jm_s2'     as score_name
    ,id                 as id           -- 外数分info表id，可关联value表info_id
    ,'jm_scorea2'       as score_key    -- 外数分键
    ,d_score            as score        -- 外数分值
    ,source             as source       -- 外数分调用方
    ,result_code        as result_code  -- 外数分结果码：1 查得分成功，0 查得分失败
    ,result_text        as result_text  -- 外数分结果文本
    ,created_time       as created_time -- 调用时间
    ,'{check_date}'    as check_date   -- 调用日期
 from fintech_jm_s2_tag
where rn = 1
)select  *  from fintech_jm_s2_user_info
"""

# 循环开始日期 结束日期
start_date = datetime.strptime('2025-11-17', '%Y-%m-%d') 
end_date = datetime.strptime('2025-11-17', '%Y-%m-%d')

end_date_str = end_date.strftime('%Y-%m-%d')

while start_date <= end_date:
    start_date_str = start_date.strftime('%Y-%m-%d')

    params = {
        'check_date': start_date_str,
        # 'month_1' : 1,
        # 'month_2' : 2,
        # 'month_3' : 3,
        # 'c_ds'    :'2024-04-11'
    }


    mysql_env.dml(run_sql.format(**params)) # 执行

    #print(run_sql.format(**params)) # 检查打印循环命令是否传如参数正确
    
    start_date += timedelta(days=1)

